package repeater

import (
	"../common"
	"bufio"
	m20 "github.com/metrics20/go-metrics20/carbon20"
	"github.com/rcrowley/goagain"
	"io"
	"net"
)

type TcpSender struct {
	output          chan *DataPointPack
	closing         chan interface{}
	stopped         chan interface{}
	datain          chan *common.DataPoint
	rc              *RepeaterController
	listenAddr      string
	batchSize       int
	readBufferSize  int
	timeWindowsSize int64
	validateLevel   m20.ValidationLevelM20
	listener        *net.TCPListener
	udpConn         *net.UDPConn
}

func (ts *TcpSender) SetOffset(offset int64) {
}
func (ts *TcpSender) Init() (err error) {
	l, err := goagain.Listener()
	if nil != err {
		laddr, err := net.ResolveTCPAddr("tcp", ts.listenAddr)
		if nil != err {
			log.Error(err.Error())
			return err
		}
		l, err = net.ListenTCP("tcp", laddr)
		if nil != err {
			log.Error(err.Error())
			return err
		}
		if nil != err {
			log.Fatalf("create tcp listener error: %v", err)
			return err
		}
		log.Infof("listening on %s", l.Addr())
	} else {
		log.Infof("resuming listening on %s", l.Addr())
		if err := goagain.Kill(); nil != err {
			log.Fatalf("kill parent process err: %v", err)
			return err
		}
	}
	ts.listener = l.(*net.TCPListener)
	return
}
func (ts *TcpSender) Serve() {
	for {
		c, err := ts.listener.AcceptTCP()
		if nil != err {
			log.Errorf("setup tcp connection, error %v", err)
			break
		}
		go ts.handle(c)
	}
}

func (ts *TcpSender) Start() {
	ts.rc.wg.Add(1)
	go ts.Serve()
	pkg := NewDataPointPack()
LOOP:
	for {
		select {
		case <-ts.closing:
			break LOOP
		case dp := <-ts.datain:
			pkg.AddDataPoint(dp, 0)
			if pkg.IsReday(ts.batchSize, ts.timeWindowsSize) {
				ts.output <- pkg
				pkg = NewDataPointPack()
			}
		}
	}
	close(ts.stopped)
}

func (ts *TcpSender) handle(c net.Conn) {
	defer c.Close()
	var addr string
	if ip := c.RemoteAddr(); ip != nil {
		addr = ip.String()
	}
	reader := bufio.NewReaderSize(c, ts.readBufferSize)
	for {
		buf, _, err := reader.ReadLine()
		if nil != err {
			if io.EOF != err {
				log.Errorf("%s tcp communication error %v", addr,err)
			}
			break
		}
		buf_copy := make([]byte, len(buf), len(buf))
		copy(buf_copy, buf)
		k, v, t, err := m20.ValidatePacket(buf_copy, m20.MediumLegacy, ts.validateLevel)
		if err != nil {
			log.Noticef("%s bad metric format %s, err %v",addr, string(buf_copy), err)
			continue
		}
		key := string(k)
		dp := common.DataPoint{key, v, t}
		ts.datain <- &dp
	}
}
func (ts *TcpSender) Stop() error {
    defer ts.rc.wg.Done()
    return ts.listener.Close()
}
func (ts *TcpSender) Name() string {
    return ts.listenAddr
}

func (ts *TcpSender) Output() <-chan *DataPointPack {
    return ts.output
}
func (ts *TcpSender) Handle(pkg *DataPointPack) error {
    return nil
}

func (ts *TcpSender) SetController(rc *RepeaterController) error {
    ts.rc = rc
    return nil
}
func (ts *TcpSender) Commit() (int64, error) {
    return 0, nil
}